package cn.org.queue.receive;

import cn.org.queue.handler.ReceiveMessageHandler;
import cn.org.queue.manage.RabbitQueueManager;
import com.rabbitmq.client.QueueingConsumer;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @Package: cn.org.queue
 * @ClassName: Consumer
 * @Description:
 * @Author: mac-pro
 * @CreateDate: 2019/2/14 上午11:32
 * @Version: 1.0
 */
@Slf4j
@NoArgsConstructor
public class RabbitConsumerMessage extends RabbitQueueManager {
    public void receiveMessage(String routingKey) {
        //声明一个临时队列 该队列会在使用完比后自动销毁
        try {
            String queueName = channel.queueDeclare().getQueue();
            //设置通道预取计数 保证一次只分发一个
            channel.basicQos(1);
            //将消息队列绑定到Exchange
            channel.queueBind(queueName, EXCHANGE_NAME, routingKey == null ? "master_routing_key" : routingKey);

            //声明消费者 - 用来缓存服务器推送过来的消息
            QueueingConsumer consumer = new QueueingConsumer(channel);
            /**
             * 监听队列，手动返回完成  - 为channel声明一个consumer，服务器会推送消息
             * 参数1：队列名称
             * 参数2：是否发送ack包，不发送ack消息会持续在服务端保存，直到收到ack。  可以通过channel.basicAck手动回复ack
             * 参数3：消费者
             */
            channel.basicConsume(queueName, false, consumer);
            //wait for the next message delivery and return it
            while (true) {
                //获取消息，如果没有消息，这一步将会一直阻塞
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                //获取消息主体数据
                byte[] body = delivery.getBody();
                //路由地址
                String tmpRoutingKey = delivery.getEnvelope().getRoutingKey();
                //对消息主体进行处理
                ReceiveMessageHandler.decodeRabbitMessage(body, tmpRoutingKey);
                //确认消息已经收到 - 回复ack包，如果不回复，消息不会在服务器删除
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        } catch (Exception e) {
            log.error("error", e);
        }
    }
}
